// MLOptimizationMapper.java
package com.taxi.analysis.ml;

import java.io.IOException;
import java.util.*;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/**
 * 智能运营优化分析Mapper
 * 功能：使用机器学习聚合技术进行运营优化分析
 * 技术：机器学习聚合技术（聚类聚合、预测聚合、分类聚合、回归聚合、集成聚合）
 */
public class MLOptimizationMapper extends Mapper<LongWritable, Text, Text, Text> {

    private Text outputKey = new Text();
    private Text outputValue = new Text();

    @Override
    public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {

        if (key.get() == 0) return;

        try {
            String line = value.toString().trim();
            if (line.isEmpty()) return;

            String[] fields = parseCSVLine(line);

            if (fields.length >= 35) {
                int pickupHour = parseInt(fields[0]);
                int pickupWeekday = parseInt(fields[1]);
                int pickupMonth = parseInt(fields[2]);
                int pickupDay = parseInt(fields[3]);
                int pickupWeek = parseInt(fields[4]);
                String pickupLocationId = fields[5].trim();
                String dropoffLocationId = fields[6].trim();
                String routeId = fields[7].trim();
                double tripDistance = parseDouble(fields[8]);
                double tripDuration = parseDouble(fields[9]);
                double avgSpeed = parseDouble(fields[10]);
                double totalAmount = parseDouble(fields[11]);
                double fareAmount = parseDouble(fields[12]);
                double tipAmount = parseDouble(fields[13]);
                int passengerCount = parseInt(fields[14]);
                double revenuePerMile = parseDouble(fields[15]);
                double revenuePerMinute = parseDouble(fields[16]);
                double tipRate = parseDouble(fields[17]);
                double passengerEfficiency = parseDouble(fields[18]);
                String timePeriod = fields[19].trim();
                String dayType = fields[20].trim();
                double weatherImpact = parseDouble(fields[21]);
                int pickupPopularity = parseInt(fields[22]);
                int dropoffPopularity = parseInt(fields[23]);
                int routePopularity = parseInt(fields[24]);
                double locationAvgFare = parseDouble(fields[25]);
                double timeAvgFare = parseDouble(fields[26]);
                double predictedDemand = parseDouble(fields[27]);
                int tripCluster = parseInt(fields[28]);
                String tripType = fields[29].trim();
                double supplyDemandRatio = parseDouble(fields[30]);
                double operationalEfficiency = parseDouble(fields[31]);
                String efficiencyLevel = fields[32].trim();
                String locationHotspot = fields[33].trim();
                int dispatchPriority = parseInt(fields[34]);
                double suggestedPriceMultiplier = parseDouble(fields[35]);
                String routeRecommendation = fields[36].trim();

                if (isValidData(pickupLocationId, dropoffLocationId)) {

                    // 机器学习聚合技术1: 聚类聚合
                    processClusteringAggregation(tripCluster, tripType, pickupHour, pickupWeekday,
                            timePeriod, dayType, tripDistance, tripDuration,
                            totalAmount, operationalEfficiency, revenuePerMile,
                            passengerEfficiency, context);

                    // 机器学习聚合技术2: 预测聚合
                    processPredictiveAggregation(pickupLocationId, pickupHour, pickupWeekday,
                            timePeriod, dayType, predictedDemand, supplyDemandRatio,
                            weatherImpact, totalAmount, operationalEfficiency,
                            dispatchPriority, context);

                    // 机器学习聚合技术3: 分类聚合
                    processClassificationAggregation(efficiencyLevel, locationHotspot, routeRecommendation,
                            tripType, timePeriod, dayType, totalAmount,
                            operationalEfficiency, revenuePerMile, tipRate,
                            suggestedPriceMultiplier, context);

                    // 机器学习聚合技术4: 回归聚合
                    processRegressionAggregation(pickupLocationId, routeId, pickupHour, pickupWeekday,
                            tripDistance, tripDuration, passengerCount, weatherImpact,
                            pickupPopularity, routePopularity, totalAmount,
                            operationalEfficiency, revenuePerMile, context);

                    // 机器学习聚合技术5: 集成聚合
                    processEnsembleAggregation(pickupLocationId, dropoffLocationId, routeId, tripCluster,
                            pickupHour, timePeriod, dayType, efficiencyLevel,
                            locationHotspot, tripType, totalAmount, operationalEfficiency,
                            predictedDemand, supplyDemandRatio, dispatchPriority,
                            suggestedPriceMultiplier, context);
                }
            }

        } catch (Exception e) {
            System.err.println("处理行时出错: " + value.toString());
        }
    }

    /**
     * 聚类聚合：基于机器学习聚类结果进行聚合
     */
    private void processClusteringAggregation(int tripCluster, String tripType, int pickupHour,
                                              int pickupWeekday, String timePeriod, String dayType,
                                              double tripDistance, double tripDuration, double totalAmount,
                                              double operationalEfficiency, double revenuePerMile,
                                              double passengerEfficiency, Context context)
            throws IOException, InterruptedException {

        // 按聚类ID聚合
        String clusterKey = String.format("CLUSTERING_ID_%d", tripCluster);
        String clusterRecord = createClusteringRecord("CLUSTER_ID", tripType, pickupHour, pickupWeekday,
                timePeriod, dayType, tripDistance, tripDuration,
                totalAmount, operationalEfficiency, revenuePerMile,
                passengerEfficiency);
        outputKey.set(clusterKey);
        outputValue.set("CLUSTERING|" + clusterRecord);
        context.write(outputKey, outputValue);

        // 按聚类类型聚合
        String typeKey = String.format("CLUSTERING_TYPE_%s", tripType);
        String typeRecord = createClusteringRecord("CLUSTER_TYPE", String.valueOf(tripCluster), pickupHour,
                pickupWeekday, timePeriod, dayType, tripDistance,
                tripDuration, totalAmount, operationalEfficiency,
                revenuePerMile, passengerEfficiency);
        outputKey.set(typeKey);
        outputValue.set("CLUSTERING|" + typeRecord);
        context.write(outputKey, outputValue);

        // 按时间-聚类聚合
        String timeClusterKey = String.format("CLUSTERING_TIME_%s_%s_%d", timePeriod, dayType, tripCluster);
        String timeClusterRecord = createClusteringRecord("TIME_CLUSTER", tripType, pickupHour,
                pickupWeekday, timePeriod, dayType, tripDistance,
                tripDuration, totalAmount, operationalEfficiency,
                revenuePerMile, passengerEfficiency);
        outputKey.set(timeClusterKey);
        outputValue.set("CLUSTERING|" + timeClusterRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 预测聚合：基于需求预测结果进行聚合
     */
    private void processPredictiveAggregation(String pickupLocationId, int pickupHour, int pickupWeekday,
                                              String timePeriod, String dayType, double predictedDemand,
                                              double supplyDemandRatio, double weatherImpact, double totalAmount,
                                              double operationalEfficiency, int dispatchPriority, Context context)
            throws IOException, InterruptedException {

        // 按预测需求等级聚合
        String demandLevel = classifyDemandLevel(predictedDemand);
        String demandKey = String.format("PREDICTIVE_DEMAND_%s", demandLevel);
        String demandRecord = createPredictiveRecord("DEMAND", pickupLocationId, pickupHour, pickupWeekday,
                timePeriod, dayType, predictedDemand, supplyDemandRatio,
                weatherImpact, totalAmount, operationalEfficiency,
                dispatchPriority);
        outputKey.set(demandKey);
        outputValue.set("PREDICTIVE|" + demandRecord);
        context.write(outputKey, outputValue);

        // 按供需比聚合
        String ratioLevel = classifySupplyDemandRatio(supplyDemandRatio);
        String ratioKey = String.format("PREDICTIVE_RATIO_%s", ratioLevel);
        String ratioRecord = createPredictiveRecord("RATIO", pickupLocationId, pickupHour, pickupWeekday,
                timePeriod, dayType, predictedDemand, supplyDemandRatio,
                weatherImpact, totalAmount, operationalEfficiency,
                dispatchPriority);
        outputKey.set(ratioKey);
        outputValue.set("PREDICTIVE|" + ratioRecord);
        context.write(outputKey, outputValue);

        // 按天气影响聚合
        String weatherLevel = classifyWeatherImpact(weatherImpact);
        String weatherKey = String.format("PREDICTIVE_WEATHER_%s", weatherLevel);
        String weatherRecord = createPredictiveRecord("WEATHER", pickupLocationId, pickupHour, pickupWeekday,
                timePeriod, dayType, predictedDemand, supplyDemandRatio,
                weatherImpact, totalAmount, operationalEfficiency,
                dispatchPriority);
        outputKey.set(weatherKey);
        outputValue.set("PREDICTIVE|" + weatherRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 分类聚合：基于分类结果进行聚合
     */
    private void processClassificationAggregation(String efficiencyLevel, String locationHotspot,
                                                  String routeRecommendation, String tripType, String timePeriod,
                                                  String dayType, double totalAmount, double operationalEfficiency,
                                                  double revenuePerMile, double tipRate, double suggestedPriceMultiplier,
                                                  Context context) throws IOException, InterruptedException {

        // 按效率等级聚合
        String efficiencyKey = String.format("CLASSIFICATION_EFFICIENCY_%s", efficiencyLevel);
        String efficiencyRecord = createClassificationRecord("EFFICIENCY", efficiencyLevel, locationHotspot,
                routeRecommendation, tripType, timePeriod, dayType,
                totalAmount, operationalEfficiency, revenuePerMile,
                tipRate, suggestedPriceMultiplier);
        outputKey.set(efficiencyKey);
        outputValue.set("CLASSIFICATION|" + efficiencyRecord);
        context.write(outputKey, outputValue);

        // 按热点等级聚合
        String hotspotKey = String.format("CLASSIFICATION_HOTSPOT_%s", locationHotspot);
        String hotspotRecord = createClassificationRecord("HOTSPOT", efficiencyLevel, locationHotspot,
                routeRecommendation, tripType, timePeriod, dayType,
                totalAmount, operationalEfficiency, revenuePerMile,
                tipRate, suggestedPriceMultiplier);
        outputKey.set(hotspotKey);
        outputValue.set("CLASSIFICATION|" + hotspotRecord);
        context.write(outputKey, outputValue);

        // 按路线推荐聚合
        String routeKey = String.format("CLASSIFICATION_ROUTE_%s", routeRecommendation);
        String routeRecord = createClassificationRecord("ROUTE", efficiencyLevel, locationHotspot,
                routeRecommendation, tripType, timePeriod, dayType,
                totalAmount, operationalEfficiency, revenuePerMile,
                tipRate, suggestedPriceMultiplier);
        outputKey.set(routeKey);
        outputValue.set("CLASSIFICATION|" + routeRecord);
        context.write(outputKey, outputValue);

        // 组合分类聚合
        String combinedKey = String.format("CLASSIFICATION_COMBINED_%s_%s_%s", efficiencyLevel, locationHotspot, routeRecommendation);
        String combinedRecord = createClassificationRecord("COMBINED", efficiencyLevel, locationHotspot,
                routeRecommendation, tripType, timePeriod, dayType,
                totalAmount, operationalEfficiency, revenuePerMile,
                tipRate, suggestedPriceMultiplier);
        outputKey.set(combinedKey);
        outputValue.set("CLASSIFICATION|" + combinedRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 回归聚合：基于回归预测进行聚合
     */
    private void processRegressionAggregation(String pickupLocationId, String routeId, int pickupHour,
                                              int pickupWeekday, double tripDistance, double tripDuration,
                                              int passengerCount, double weatherImpact, int pickupPopularity,
                                              int routePopularity, double totalAmount, double operationalEfficiency,
                                              double revenuePerMile, Context context) throws IOException, InterruptedException {

        // 收入回归聚合
        String revenueKey = String.format("REGRESSION_REVENUE_%s_%d_%d", pickupLocationId, pickupHour, pickupWeekday);
        String revenueRecord = createRegressionRecord("REVENUE", pickupLocationId, routeId, pickupHour,
                pickupWeekday, tripDistance, tripDuration, passengerCount,
                weatherImpact, pickupPopularity, routePopularity,
                totalAmount, operationalEfficiency, revenuePerMile);
        outputKey.set(revenueKey);
        outputValue.set("REGRESSION|" + revenueRecord);
        context.write(outputKey, outputValue);

        // 效率回归聚合
        String efficiencyKey = String.format("REGRESSION_EFFICIENCY_%s", routeId);
        String efficiencyRecord = createRegressionRecord("EFFICIENCY", pickupLocationId, routeId, pickupHour,
                pickupWeekday, tripDistance, tripDuration, passengerCount,
                weatherImpact, pickupPopularity, routePopularity,
                totalAmount, operationalEfficiency, revenuePerMile);
        outputKey.set(efficiencyKey);
        outputValue.set("REGRESSION|" + efficiencyRecord);
        context.write(outputKey, outputValue);

        // 时间回归聚合
        String timeKey = String.format("REGRESSION_TIME_%d_%d", pickupHour, pickupWeekday);
        String timeRecord = createRegressionRecord("TIME", pickupLocationId, routeId, pickupHour,
                pickupWeekday, tripDistance, tripDuration, passengerCount,
                weatherImpact, pickupPopularity, routePopularity,
                totalAmount, operationalEfficiency, revenuePerMile);
        outputKey.set(timeKey);
        outputValue.set("REGRESSION|" + timeRecord);
        context.write(outputKey, outputValue);

        // 距离回归聚合
        String distanceCategory = classifyDistance(tripDistance);
        String distanceKey = String.format("REGRESSION_DISTANCE_%s", distanceCategory);
        String distanceRecord = createRegressionRecord("DISTANCE", pickupLocationId, routeId, pickupHour,
                pickupWeekday, tripDistance, tripDuration, passengerCount,
                weatherImpact, pickupPopularity, routePopularity,
                totalAmount, operationalEfficiency, revenuePerMile);
        outputKey.set(distanceKey);
        outputValue.set("REGRESSION|" + distanceRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 集成聚合：综合多种机器学习技术进行聚合
     */
    private void processEnsembleAggregation(String pickupLocationId, String dropoffLocationId, String routeId,
                                            int tripCluster, int pickupHour, String timePeriod, String dayType,
                                            String efficiencyLevel, String locationHotspot, String tripType,
                                            double totalAmount, double operationalEfficiency, double predictedDemand,
                                            double supplyDemandRatio, int dispatchPriority, double suggestedPriceMultiplier,
                                            Context context) throws IOException, InterruptedException {

        // 综合优化聚合
        String optimizationKey = String.format("ENSEMBLE_OPTIMIZATION_%s_%s", efficiencyLevel, locationHotspot);
        String optimizationRecord = createEnsembleRecord("OPTIMIZATION", pickupLocationId, dropoffLocationId,
                routeId, tripCluster, pickupHour, timePeriod, dayType,
                efficiencyLevel, locationHotspot, tripType, totalAmount,
                operationalEfficiency, predictedDemand, supplyDemandRatio,
                dispatchPriority, suggestedPriceMultiplier);
        outputKey.set(optimizationKey);
        outputValue.set("ENSEMBLE|" + optimizationRecord);
        context.write(outputKey, outputValue);

        // 智能调度聚合
        String dispatchKey = String.format("ENSEMBLE_DISPATCH_%s_%s_%s", timePeriod, dayType, locationHotspot);
        String dispatchRecord = createEnsembleRecord("DISPATCH", pickupLocationId, dropoffLocationId, routeId,
                tripCluster, pickupHour, timePeriod, dayType, efficiencyLevel,
                locationHotspot, tripType, totalAmount, operationalEfficiency,
                predictedDemand, supplyDemandRatio, dispatchPriority,
                suggestedPriceMultiplier);
        outputKey.set(dispatchKey);
        outputValue.set("ENSEMBLE|" + dispatchRecord);
        context.write(outputKey, outputValue);

        // 收益最大化聚合
        String revenueKey = String.format("ENSEMBLE_REVENUE_%s_%d", tripType, tripCluster);
        String revenueRecord = createEnsembleRecord("REVENUE", pickupLocationId, dropoffLocationId, routeId,
                tripCluster, pickupHour, timePeriod, dayType, efficiencyLevel,
                locationHotspot, tripType, totalAmount, operationalEfficiency,
                predictedDemand, supplyDemandRatio, dispatchPriority,
                suggestedPriceMultiplier);
        outputKey.set(revenueKey);
        outputValue.set("ENSEMBLE|" + revenueRecord);
        context.write(outputKey, outputValue);

        // 全局集成聚合
        String globalKey = "ENSEMBLE_GLOBAL";
        String globalRecord = createEnsembleRecord("GLOBAL", pickupLocationId, dropoffLocationId, routeId,
                tripCluster, pickupHour, timePeriod, dayType, efficiencyLevel,
                locationHotspot, tripType, totalAmount, operationalEfficiency,
                predictedDemand, supplyDemandRatio, dispatchPriority,
                suggestedPriceMultiplier);
        outputKey.set(globalKey);
        outputValue.set("ENSEMBLE|" + globalRecord);
        context.write(outputKey, outputValue);
    }

    // 辅助方法
    private String classifyDemandLevel(double predictedDemand) {
        if (predictedDemand >= 100) return "高需求";
        else if (predictedDemand >= 50) return "中需求";
        else return "低需求";
    }

    private String classifySupplyDemandRatio(double ratio) {
        if (ratio < 0.8) return "供不应求";
        else if (ratio > 1.2) return "供过于求";
        else return "供需平衡";
    }

    private String classifyWeatherImpact(double impact) {
        if (impact < 0.6) return "恶劣天气";
        else if (impact > 0.8) return "良好天气";
        else return "一般天气";
    }

    private String classifyDistance(double distance) {
        if (distance <= 2) return "短途";
        else if (distance <= 10) return "中途";
        else return "长途";
    }

    // 记录创建方法
    private String createClusteringRecord(String type, String param1, int pickupHour, int pickupWeekday,
                                          String timePeriod, String dayType, double tripDistance, double tripDuration,
                                          double totalAmount, double operationalEfficiency, double revenuePerMile,
                                          double passengerEfficiency) {
        return String.format("%s|%s|%d|%d|%s|%s|%.2f|%.2f|%.2f|%.2f|%.2f|%.4f",
                type, param1, pickupHour, pickupWeekday, timePeriod, dayType,
                tripDistance, tripDuration, totalAmount, operationalEfficiency,
                revenuePerMile, passengerEfficiency);
    }

    private String createPredictiveRecord(String type, String pickupLocationId, int pickupHour, int pickupWeekday,
                                          String timePeriod, String dayType, double predictedDemand,
                                          double supplyDemandRatio, double weatherImpact, double totalAmount,
                                          double operationalEfficiency, int dispatchPriority) {
        return String.format("%s|%s|%d|%d|%s|%s|%.2f|%.3f|%.3f|%.2f|%.2f|%d",
                type, pickupLocationId, pickupHour, pickupWeekday, timePeriod, dayType,
                predictedDemand, supplyDemandRatio, weatherImpact, totalAmount,
                operationalEfficiency, dispatchPriority);
    }

    private String createClassificationRecord(String type, String efficiencyLevel, String locationHotspot,
                                              String routeRecommendation, String tripType, String timePeriod,
                                              String dayType, double totalAmount, double operationalEfficiency,
                                              double revenuePerMile, double tipRate, double suggestedPriceMultiplier) {
        return String.format("%s|%s|%s|%s|%s|%s|%s|%.2f|%.2f|%.2f|%.3f|%.2f",
                type, efficiencyLevel, locationHotspot, routeRecommendation, tripType,
                timePeriod, dayType, totalAmount, operationalEfficiency, revenuePerMile,
                tipRate, suggestedPriceMultiplier);
    }

    private String createRegressionRecord(String type, String pickupLocationId, String routeId, int pickupHour,
                                          int pickupWeekday, double tripDistance, double tripDuration,
                                          int passengerCount, double weatherImpact, int pickupPopularity,
                                          int routePopularity, double totalAmount, double operationalEfficiency,
                                          double revenuePerMile) {
        return String.format("%s|%s|%s|%d|%d|%.2f|%.2f|%d|%.3f|%d|%d|%.2f|%.2f|%.2f",
                type, pickupLocationId, routeId, pickupHour, pickupWeekday, tripDistance,
                tripDuration, passengerCount, weatherImpact, pickupPopularity, routePopularity,
                totalAmount, operationalEfficiency, revenuePerMile);
    }

    private String createEnsembleRecord(String type, String pickupLocationId, String dropoffLocationId,
                                        String routeId, int tripCluster, int pickupHour, String timePeriod,
                                        String dayType, String efficiencyLevel, String locationHotspot,
                                        String tripType, double totalAmount, double operationalEfficiency,
                                        double predictedDemand, double supplyDemandRatio, int dispatchPriority,
                                        double suggestedPriceMultiplier) {
        return String.format("%s|%s|%s|%s|%d|%d|%s|%s|%s|%s|%s|%.2f|%.2f|%.2f|%.3f|%d|%.2f",
                type, pickupLocationId, dropoffLocationId, routeId, tripCluster, pickupHour,
                timePeriod, dayType, efficiencyLevel, locationHotspot, tripType, totalAmount,
                operationalEfficiency, predictedDemand, supplyDemandRatio, dispatchPriority,
                suggestedPriceMultiplier);
    }

    private boolean isValidData(String pickupLocationId, String dropoffLocationId) {
        return !pickupLocationId.isEmpty() && !dropoffLocationId.isEmpty() &&
                !pickupLocationId.equals("null") && !dropoffLocationId.equals("null");
    }

    private String[] parseCSVLine(String line) {
        if (line.contains("\"")) {
            return parseQuotedCSV(line);
        } else {
            return line.split(",");
        }
    }

    private String[] parseQuotedCSV(String line) {
        List<String> fields = new ArrayList<>();
        boolean inQuotes = false;
        StringBuilder currentField = new StringBuilder();

        for (int i = 0; i < line.length(); i++) {
            char c = line.charAt(i);

            if (c == '"') {
                inQuotes = !inQuotes;
            } else if (c == ',' && !inQuotes) {
                fields.add(currentField.toString());
                currentField = new StringBuilder();
            } else {
                currentField.append(c);
            }
        }
        fields.add(currentField.toString());

        return fields.toArray(new String[0]);
    }

    private double parseDouble(String str) {
        try {
            return Double.parseDouble(str.trim());
        } catch (NumberFormatException e) {
            return 0.0;
        }
    }

    private int parseInt(String str) {
        try {
            return Integer.parseInt(str.trim());
        } catch (NumberFormatException e) {
            return 0;
        }
    }
}
